package com.galeno.day08;

import com.galeno.FlinkUtils;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.util.Properties;

/**
 * @author galeno
 * @Title:
 * @Description:
 * @date 2021/10/2517:24
 */
public class KafkaToRedisWordCountAggregate {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(10000);
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", FlinkUtils.kafkaBootstrapAddr);
        properties.setProperty("group.id", "test1");
        properties.setProperty("auto.offset.reset", "earliest");
        FlinkKafkaConsumer<String> flinkKafkaConsumer = new FlinkKafkaConsumer<>("wc", new SimpleStringSchema(), properties);
        flinkKafkaConsumer.setCommitOffsetsOnCheckpoints(false);
        DataStreamSource<String> lines = env.addSource(flinkKafkaConsumer);
        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAnd1 = lines.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
                String[] words = value.split("\\s+");
                for (String word : words) {
                    out.collect(Tuple2.of(word, 1));
                }
            }
        });
        KeyedStream<Tuple2<String, Integer>, String> keyedStream = wordAnd1.keyBy(x -> x.f0);
        WindowedStream<Tuple2<String, Integer>, String, TimeWindow> window = keyedStream.window(TumblingProcessingTimeWindows.of(Time.seconds(10)));
      //  window.aggregate();
        SingleOutputStreamOperator<Tuple2<String, Integer>> aggregate = window.aggregate(new MyAgg(), new MyAggWin());
        aggregate.print();
        env.execute();


    }
    private static class MyAgg implements AggregateFunction<Tuple2<String,Integer>,Integer,Integer>{

        @Override
        public Integer createAccumulator() {
            return 0;
        }

        @Override
        public Integer add(Tuple2<String, Integer> value, Integer accumulator) {
           return value.f1+accumulator;


        }

        @Override
        public Integer getResult(Integer accumulator) {
           return accumulator;

        }

        /**
         * session窗口才会调用
         * @param a
         * @param b
         * @return
         */
        @Override
        public Integer merge(Integer a, Integer b) {
            return null;
        }
    }
    private static class MyAggWin extends ProcessWindowFunction<Integer,Tuple2<String,Integer>,String,TimeWindow>{
        private transient ValueState<Integer> valueState;

        @Override
        public void open(Configuration parameters) throws Exception {
            ValueStateDescriptor<Integer> valueStateDescriptor = new ValueStateDescriptor<>("count-state", Integer.class);
            valueState= getRuntimeContext().getState(valueStateDescriptor);
        }



        @Override
        public void process(String s, Context context, Iterable<Integer> elements, Collector<Tuple2<String, Integer>> out) throws Exception {
            int windowCount = elements.iterator().next();
            Integer historyvalue = valueState.value();
            if (historyvalue==null){
                historyvalue=0;
            }
            historyvalue+=windowCount;
            valueState.update(historyvalue);
            out.collect(Tuple2.of(s,historyvalue));
        }
    }
}
